package com.hexinfo.dmpro.dev.data.transfer.handle.impl;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson.JSON;
import com.hexinfo.dmpro.data.transfer.common.common.manager.AgentManager;
import com.hexinfo.dmpro.dev.data.transfer.common.enums.DevDataTransferExportTaskStatusEnum;
import com.hexinfo.dmpro.dev.data.transfer.common.enums.DevDataTransferListStatusEnum;
import com.hexinfo.dmpro.dev.data.transfer.handle.IDevDataExportHandler;
import com.hexinfo.dmpro.dev.data.transfer.model.DevDataExportList;
import com.hexinfo.dmpro.dev.data.transfer.model.DevDataExportTask;
import com.hexinfo.dmpro.dev.data.transfer.service.IDevDataExportListService;
import com.hexinfo.dmpro.dev.data.transfer.service.IDevDataExportTaskService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.util.Arrays;
import java.util.List;

@Slf4j
@Service
public class DevDataExportHandler implements IDevDataExportHandler {

    @Autowired
    private IDevDataExportListService dataExportListService;

    @Autowired
    private IDevDataExportTaskService devDataExportTaskService;

    @Autowired
    private IDevDataExportListService devDataExportListService;

    @Autowired
    private AgentManager agentManager;

    @Value("${conn.pool.hive.target.url}")
    private String hiveJdbcUrl;

    @Value("${conn.pool.hive.target.username}")
    private String hiveUserName;

    @Value("${conn.pool.hive.target.password}")
    private String hivePassword;

    @Value("${dev-data-ftp.server}")
    private String ftpServer;

    @Value("${dev-data-ftp.port}")
    private Integer ftpPort;

    @Value("${dev-data-ftp.user}")
    private String ftpUserName;

    @Value("${dev-data-ftp.password}")
    private String ftpPassword;

    @Value("${nameNode.target.nameNodeIp}")
    private String hdfsNameNodeHosts;

    /*
     * @Method: execute <br>
     * @Param: [hdfsUrl, devDataExportList] <br>
     * @Return: void <br>
     * @Description：执行开发数据同步任务<br>
     * @Author： wz.li<br>
     * @Date： 2023/11/6 19:05 <br>
     * @Version： V2.0.2<br>
     */
    @Override
    @Async("devDataTransferAsyncTaskExecutor")
    public void executeExportTask(DevDataExportList devDataExportList) {
        try {
            // 判断任务状态是否中止
            if(exportTaskIsAbolish(devDataExportList.getTaskCode())){
                // 修改清单状态
                abolishExportList(devDataExportList);
            }else if (exportTaskIsRun(devDataExportList.getTaskCode())) {
                // ① 补充导出清单配置信息
                supplementDevDataExportList(devDataExportList);
                // ② 执行开发数据同步任务
                executeTask(devDataExportList);
            } else {
                log.warn(StrUtil.format("表【{}】开发数据导出暂停执行", devDataExportList.getDbAndTableName()));
                ThreadUtil.sleep(1000);
                dataExportListService.restart(devDataExportList.getId());
            }
        } catch (Exception e) {
            log.error(StrUtil.format("表【{}】开发数据导出异常, 异常信息: {}", devDataExportList.getDbAndTableName(), e.getMessage()), e);
            updateDevDataExportListToFail(devDataExportList, e.getMessage());
        }
    }

    /**
     * @Method exportTaskIsAbolish
     * @Param
     * @param taskCode
     * @Return boolean
     * @Description 判断任务是否中止执行
     * @Author Wang zhihao
     * @Date 2024/1/3 19:06
     * @Version V1.0
     */
    private boolean exportTaskIsAbolish(String taskCode) {
        DevDataExportTask devDataExportTask = devDataExportTaskService.queryByTaskCode(taskCode);
        return DevDataTransferExportTaskStatusEnum.abolish.getValue().equals(devDataExportTask.getTaskStatus());
    }


    /**
     * @Method abolishExportList
     * @Param
     * @param devDataExportList
     * @Return void
     * @Description 取消指定的开发数据导出表清单
     * @Author Wang zhihao
     * @Date 2024/1/3 18:53
     * @Version V1.0
     */
    private void abolishExportList(DevDataExportList devDataExportList) {
        if (DevDataTransferListStatusEnum.TODO.getValue().equals(devDataExportList.getExportStatus())) {
            devDataExportList.setExportStatus(DevDataTransferListStatusEnum.ABOLISH.getValue());
            devDataExportListService.updateById(devDataExportList);
        }
    }

    /*
     * @Method: exportTaskIsRun <br>
     * @Param: [taskCode] <br>
     * @Return: boolean <br>
     * @Description：判断任务是否暂停执行<br>
     * @Author： wz.li<br>
     * @Date： 2023/12/5 16:52 <br>
     * @Version： V2.0.2<br>
     */
    private boolean exportTaskIsRun(String taskCode) {
        DevDataExportTask devDataExportTask = devDataExportTaskService.queryByTaskCode(taskCode);
        return !DevDataTransferExportTaskStatusEnum.suspend.getValue().equals(devDataExportTask.getTaskStatus());
    }

    /*
     * @Method: executeTask <br>
     * @Param: [devDataExportList] <br>
     * @Return: void <br>
     * @Description：向AGETN发送导出命令<br>
     * @Author： wz.li<br>
     * @Date： 2023/12/5 16:53 <br>
     * @Version： V2.0.2<br>
     */
    private synchronized void executeTask(DevDataExportList devDataExportList) {
        try {
            // ③ 获取可用Agent
            String agentKey = getAvailableAgent();
            devDataExportList.setAgentKey(agentKey);
            // ④ 发送Agent开发数据导出任务
            sendAgentExport(devDataExportList, agentKey);
            // ⑤ 更新导出清单状态为执行中
            updateDevDataExportListToRunning(devDataExportList, agentKey);
            // ⑥ AGNET新增执行任务
            agentManager.addExecuteTask(agentKey);
        } catch (Exception e) {
            log.warn(StrUtil.format("表【{}】开发数据导出发送失败: {}", devDataExportList.getDbAndTableName(), e.getMessage()));
            ThreadUtil.sleep(10000);
            dataExportListService.restart(devDataExportList.getId());
        }
    }

    /*
     * @Method: updateDevDataExportListToRunning <br>
     * @Param: [devDataExportList] <br>
     * @Return: void <br>
     * @Description：更新开发数据导出为执行中<br>
     * @Author： wz.li<br>
     * @Date： 2023/11/6 19:04 <br>
     * @Version： V2.0.2<br>
     */
    private void updateDevDataExportListToRunning(DevDataExportList devDataExportList, String agentKey) {
        devDataExportList.setExportStatus(DevDataTransferListStatusEnum.RUNNING.getValue());
        devDataExportList.setStartTime(DateUtil.date());
        devDataExportList.setAgentKey(agentKey);
        dataExportListService.updateById(devDataExportList);
    }

    /*
     * @Method: getAvailableAgent <br>
     * @Param: [] <br>
     * @Return: com.hexinfo.dmpro.data.transfer.common.model.AgentInfo <br>
     * @Description：获取可用的数据迁移服务<br>
     * @Author： wz.li<br>
     * @Date： 2023/11/6 19:04 <br>
     * @Version： V2.0.2<br>
     */
    private String getAvailableAgent() {
        String availableAgent = agentManager.getAvailableAgent();
        if (StrUtil.isBlank(availableAgent)) {
            throw new RuntimeException("没有可用的执行AGENT，请等待AGENT释放资源！");
        }
        return availableAgent;
    }

    /*
     * @Method: supplementDevDataExportList <br>
     * @Param: [devDataExportList] <br>
     * @Return: void <br>
     * @Description：补充导出清单配置信息<br>
     * @Author： wz.li<br>
     * @Date： 2023/11/12 14:43 <br>
     * @Version： V2.0.2<br>
     */
    private void supplementDevDataExportList(DevDataExportList devDataExportList) {
        devDataExportList.setHdfsTmpPath("/tmp/dmpro");
        devDataExportList.setFtpHost(ftpServer);
        devDataExportList.setFtpPort(ftpPort);
        devDataExportList.setFtpUserName(ftpUserName);
        devDataExportList.setFtpPassword(ftpPassword);
        devDataExportList.setHiveJdbcUrl(hiveJdbcUrl);
        devDataExportList.setHiveUserName(hiveUserName);
        devDataExportList.setHivePassword(hivePassword);
        devDataExportList.setHdfsNameNodeHosts(Arrays.asList(StrUtil.split(hdfsNameNodeHosts, ",")));
        devDataExportList.setHdfsPort(8020);
    }

    /*
     * @Method: sendAgentExport <br>
     * @Param: [hiveDataExportModel, agentInfo] <br>
     * @Return: void <br>
     * @Description：发送开发数据导出任务<br>
     * @Author： wz.li<br>
     * @Date： 2023/11/6 19:03 <br>
     * @Version： V2.0.2<br>
     */
    private void sendAgentExport(DevDataExportList devDataExportList, String agentKey) {
        try {
            String httpUrl = StrUtil.format("http://{}/api/hive/export", agentKey);
            final HttpRequest post = HttpUtil.createPost(httpUrl);
            String body = JSON.toJSONString(devDataExportList);
            log.info("表【{}】开发数据导出请求参数: {}", devDataExportList.getDbAndTableName(), body);
            final HttpResponse execute = post.body(body).execute();
            if (200 != execute.getStatus()) {
                throw new RuntimeException("HTTP请求失败: " + execute.body());
            }
            log.info("表【{}】开发数据导出请求成功！(●'◡'●)", devDataExportList.getDbAndTableName());
        } catch (Exception e) {
            throw new RuntimeException(StrUtil.format("发送HTTP请求失败, 失败原因{}", e.getMessage()), e);
        }
    }

    /*
     * @Method: updateDevDataExportListToFail <br>
     * @Param: [devDataExportList] <br>
     * @Return: void <br>
     * @Description：开发数据导出执行失败更新<br>
     * @Author： wz.li<br>
     * @Date： 2023/11/6 19:03 <br>
     * @Version： V2.0.2<br>
     */
    private void updateDevDataExportListToFail(DevDataExportList devDataExportList, String errorMessage) {
        devDataExportList.setExportStatus(DevDataTransferListStatusEnum.FAIL.getValue());
        devDataExportList.setErrorMessage(errorMessage);
        devDataExportList.setEndTime(DateUtil.date());
        dataExportListService.updateById(devDataExportList);
        devDataExportTaskService.modifyFailNumByTaskCode(devDataExportList.getTaskCode());
    }
}
